package slidingwindow

import (
	"sync"
	"sync/atomic"
	"time"
)

type WindowTime struct {
	timeLength    int32         // 时间长度,单位秒
	bucketNum     int32         // 桶数量
	buckets       []int32       // 滑窗
	index         int32         // 待更新桶下标，循环移动
	num           int32         // 滑窗内事件数量
	bucketFilling int32         // 正在填充的桶
	isRun         int32         // 是否在运行 0未运行,1运行
	invalid       int32         // 输出是否有效，0 无效，1有效
	closeChan     chan struct{} // 运行关闭的通知通道
	lock          sync.Mutex    // 主要时对buckets加锁
}

func NewWindowTime() ISlidingWindows {
	return &WindowTime{
		timeLength: 1,
		bucketNum:  10,
		closeChan:  make(chan struct{}, 1),
	}
}

func (s *WindowTime) SetUp(timeLength int32, bucketNum int32) error {
	if atomic.LoadInt32(&s.isRun) == 1 {
		return RunIngError
	} else {
		atomic.StoreInt32(&s.timeLength, timeLength)
		atomic.StoreInt32(&s.bucketNum, bucketNum)
		return nil
	}
}

func (s *WindowTime) Event() error {
	if atomic.LoadInt32(&s.isRun) == 0 {
		return NoRunError
	} else {
		atomic.AddInt32(&s.bucketFilling, 1)
		return nil
	}
}

func (s *WindowTime) Output() (int32, float32, error) {
	if atomic.LoadInt32(&s.invalid) == 0 {
		return 0, 0, InvalidError
	} else {
		ti := atomic.LoadInt32(&s.timeLength)
		num := atomic.LoadInt32(&s.num)
		return num, float32(num) / float32(ti), nil
	}
}

func (s *WindowTime) Run() error {
	if atomic.LoadInt32(&s.isRun) == 1 {
		return RunIngError
	}
	s.lock.Lock()
	defer s.lock.Unlock()
	s.buckets = make([]int32, s.bucketNum)
	s.index = 0

	atomic.StoreInt32(&s.num, 0)
	atomic.StoreInt32(&s.bucketFilling, 0)
	atomic.StoreInt32(&s.isRun, 1)
	atomic.StoreInt32(&s.invalid, 1)
	timeLength := atomic.LoadInt32(&s.timeLength)
	bucketNum := atomic.LoadInt32(&s.bucketNum)
	ti := float32(timeLength) / float32(bucketNum) * 1000 // 毫秒
	go func() {
		// 做定时
		timeout := make(chan bool, 1)
		timeoutFunc := func() {
			time.Sleep(time.Duration(ti) * time.Millisecond) // 指定超时时长
			timeout <- true
		}
		go timeoutFunc()
		for {
			select {
			case <-timeout:
				s.lock.Lock()
				// 剔除原始数据并从总请求数中减去,插入新聚合桶数据。
				atomic.AddInt32(&s.num, -s.buckets[s.index])
				s.buckets[s.index] = atomic.LoadInt32(&s.bucketFilling)
				// 新桶清洗成0
				atomic.StoreInt32(&s.bucketFilling, 0)
				// 更新总请求数，加上新增的桶的数值
				atomic.AddInt32(&s.num, s.buckets[s.index])
				// 更新桶队列下角标
				count := atomic.LoadInt32(&s.bucketNum)
				s.index++
				if s.index >= count {
					s.index = 0
				}
				s.lock.Unlock()
				// 继续做超时
				go timeoutFunc()
			case <-s.closeChan:
				// 结束运行状态，并退出groutine
				atomic.StoreInt32(&s.isRun, 0)
				break
			}
		}
	}()
	return nil
}

func (s *WindowTime) Stop() error {
	if atomic.LoadInt32(&s.isRun) == 1 {
		close(s.closeChan) //关闭运行通道
		return nil
	} else {
		return NoRunError
	}
}
